package a_o2odata_deal

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.spark._

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2019/12/7 15:06
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object modify_2018 {

  def pricedeal(priceText: String): String = {
    var price = "-1"
    if (!priceText.equals("-1")) {
      if (priceText.contains("-")) {
        price = priceText.split("-")(0)
      } else {
        price = priceText
      }
    } else {
      price = "-1"
    }
    price
  }


  def min(price1:String,price2:String):Double={
    val p1 = price1.toDouble
    val p2 = price2.toDouble

    math.min(p1,p2)
  }

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.debug.maxToStringFields", "500")
    conf.setMaster("local[*]")
    conf.set("es.nodes", "192.168.1.157")
    conf.set("es.port", "9200")
    conf.set("cluster.name", "O2OElastic")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.sql.caseSensitive", "true")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    sc.setLogLevel("WARN")
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")

    sqlContext.udf.register("min",min _)
    sqlContext.udf.register("pricedeal", pricedeal _)

    //拉取数据
   /* val rdd = sc.esJsonRDD("2018_tmall/tmall_2018_11").values
    val data = sqlContext.read.json(rdd)

    data.repartition(10).write.json("s3a://o2o-sourcedata/obs-source-2019/shuang11/normalSales/tmall/month11_data_2018_all")*/


    //计算  取出关联到的
    /*val source_data = sqlContext.read.json("s3a://o2o-sourcedata/obs-source-2019/shuang11/normalSales/tmall/month11_data_2018_all")

    val price_data = sqlContext.read.json("s3a://o2o-sourcedata/obs-source-2019/shuang11/normalSales/tmall/tmall_2018data_path").where("priceText>0")

    source_data.createOrReplaceTempView("source_data")
    price_data.createOrReplaceTempView("price_data")

    val result_join = sqlContext.sql(
      s"""
         |select
         |*,
         |cast(price as double)*cast(sellCount as bigint) as sales
         |from
         |(select
         |t1.*,
         |min(pricedeal(t1.priceText),pricedeal(t2.priceText)) as price
         |from source_data t1
         |inner join
         |price_data t2
         |on t1.good_id=t2.good_id)
       """.stripMargin).drop("priceText","salesAmount")
      .withColumnRenamed("price","priceText")
      .withColumnRenamed("sales","salesAmount")

    result_join.repartition(4).write.json("s3a://o2o-sourcedata/obs-source-2019/shuang11/normalSales/tmall/month11_data_2018_all_join_v1")


    import org.apache.spark.sql.functions._
    result_join.agg(sum("salesAmount")).show(false)*/


    //入库
    val result = sqlContext.read.json("s3a://o2o-sourcedata/obs-source-2019/shuang11/normalSales/tmall/month11_data_2018_all_join_v1")

    val value = result.toJSON.rdd.map(line => {
      val lines = JSON.parseObject(line)
      val evaluates = lines.getOrDefault("evaluates", "-1").toString
      val Base_Info = lines.getOrDefault("Base_Info", "-1").toString
      var promotion_price = lines.getOrDefault("promotion_price","-1").toString
      var original_cost = lines.getOrDefault("original_cost","-1").toString
      val priceText = lines.get("priceText").toString
      if (promotion_price.contains("?") | promotion_price.contains("？")){
        promotion_price = priceText
      }
      if (original_cost.contains("?") | original_cost.contains("？")){
        original_cost = priceText
      }
      var ev = "-1"
      var str = new JSONObject
      if (evaluates.equals("-1")) {
        ev = "{\"fuyi\":\"-1\"}"
      } else {
        str = JSON.parseObject(evaluates)
      }
      if (!ev.contains("fuyi")) {
        lines.put("evaluates", str)
      } else {
        val evs = JSON.parseObject(ev)
        lines.put("evaluates", evs)
      }
      lines.put("Base_Info",Base_Info)
      lines.put("promotion_price",promotion_price)
      lines.put("original_cost",original_cost)
      lines
    })
    //原始代码
    value.saveToEs(s"2018_tmall/tmall_2018_11", Map("es.mapping.id" -> "good_id"))
  }

}
